package org.yun.octopus.redis.core;

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.yun.octopus.base.config.Cfg;
import org.yun.octopus.base.spi.IMarkCache;
import org.yun.octopus.base.util.SystemTimer;
import org.yun.octopus.redis.async.RedisAsyncMark;
import org.yun.octopus.redis.util.ORedisUtil;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

import static org.yun.octopus.base.core.RollbackFactory.RELIABLE_FLAGS;
import static org.yun.octopus.base.core.RollbackFactory.STEP_RESULT_MAP;
import static org.yun.octopus.base.core.RollbackFactory.UNRELIABLE_FLAGS;
import static org.yun.octopus.base.util.CommonUtil.initMap;
import static org.yun.octopus.base.util.CommonUtil.joinMap;


/**
 * @author liyunfeng31
 */
@Component
public class RedisSyncMark implements IMarkCache {

    @Resource
    private ORedisUtil redisUtil;

    /**
     * 缓存打标
     * @param group 业务tag
     * @param bizNo 业务编号
     */
    @Override
    public boolean syncSave(String group, String bizNo) {
        return save( group, bizNo,0,0,null) ;
    }

    /**
     * 缓存打标
     *
     * @param group  业务tag
     * @param bizNo  业务编号
     * @param stepNo 步骤编号
     * @param code   步骤结果标记
     */
    @Override
    public boolean save(String group, String bizNo, int stepNo, int code) {
        return save(group,bizNo,stepNo,code,null);
    }


    @Override
    public boolean save(String group, String bizNo, int stepNo, int code, String childNo) {

        if(Cfg.markAsync && stepNo != 0){
           return RedisAsyncMark.offer(group, bizNo, stepNo, code, childNo);
        }

        String key = group+"_"+bizNo;
        if(redisUtil.zAdd(group, bizNo, SystemTimer.currentTime())){
            redisUtil.hPutAll(key, stepNo == 0 ? STEP_RESULT_MAP : initMap(stepNo,code));
            return true;
        }

        if(childNo != null){
            redisUtil.hPutAll(key, joinMap(stepNo,code,childNo));
        }else{
            redisUtil.hPut(key, stepNo,code);
        }
        return true;
    }


    @Override
    public List<Integer> stepNos(String group, String bizNo) {
        return redisUtil.hGetAll(group + "_" + bizNo).entrySet().stream()
                .filter(map -> contains(map.getValue()))
                .map(map->Integer.parseInt(map.getKey().toString()))
                .collect(Collectors.toList());
    }


    @Override
    public boolean clearMark(String group, String bizNo) {

        if(Cfg.markAsync){
           return RedisAsyncMark.offer(group, bizNo);
        }

        redisUtil.del(group+"_"+bizNo);
        redisUtil.zRemoveMember(group,bizNo);
        return true;
    }


    @Override
    public List<Object> abnormalNos(String group, long st, long et, int limit) {
        Set<String> set = redisUtil.zRangeByScore(group, st, et, 0, limit);
        return set == null ? null : new ArrayList<>(set);
    }

    @Override
    public String childBizNo(String group, String ctxBizNo, int no) {
        Object val = redisUtil.hGet(group + "_" + ctxBizNo, no + "No");
        return val.toString();
    }


    private boolean contains(Object val){
        Set<String> set  = Cfg.markAsync ? UNRELIABLE_FLAGS : RELIABLE_FLAGS;
        return set.contains(val.toString());
    }
}
